package kfk

import (
	"fmt"
	"github.com/Shopify/sarama"
	"kafka/common/handlers/conf"
	"strings"
)

func ConsumerRun() {
	fmt.Println("run consumer_test")

	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Version = sarama.V0_11_0_2

	// consumer
	address := strings.Split(conf.Viper.GetString("kafka.address"), ",")
	consumer, err := sarama.NewConsumer(address, config)
	if err != nil {
		fmt.Printf("consumer_test create consumer error %s\n", err.Error())
		return
	}

	//defer consumer.Close()

	partition_consumer0, err := consumer.ConsumePartition("topic001", 0, sarama.OffsetOldest)
	if err != nil {
		fmt.Printf("try create partition_consumer0 error %s\n", err.Error())
		return
	}

	go func() {
		defer partition_consumer0.Close()
		for {
			fmt.Println("listen consumer_test0 >>>>>>")
			select {
			case msg := <-partition_consumer0.Messages():
				fmt.Printf("msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
					msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
			case err := <-partition_consumer0.Errors():
				fmt.Printf("err :%s\n", err.Error())
			}
		}
	}()

	partition_consumer1, err := consumer.ConsumePartition("topic001", 1, sarama.OffsetOldest)
	if err != nil {
		fmt.Printf("try create partition_consumer1 error %s\n", err.Error())
		return
	}
	go func() {
		defer partition_consumer1.Close()
		for {
			fmt.Println("listen consumer_test1 >>>>>>")
			select {
			case msg := <-partition_consumer1.Messages():
				fmt.Printf("msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
					msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
			case err := <-partition_consumer1.Errors():
				fmt.Printf("err :%s\n", err.Error())
			}
		}
	}()

	partition_consumer2, err := consumer.ConsumePartition("topic001", 2, sarama.OffsetOldest)
	if err != nil {
		fmt.Printf("try create partition_consumer2 error %s\n", err.Error())
		return
	}
	go func() {
		defer partition_consumer2.Close()
		for {
			fmt.Println("listen consumer_test2 >>>>>>")
			select {
			case msg := <-partition_consumer2.Messages():
				fmt.Printf("msg offset: %d, partition: %d, timestamp: %s, value: %s\n",
					msg.Offset, msg.Partition, msg.Timestamp.String(), string(msg.Value))
			case err := <-partition_consumer2.Errors():
				fmt.Printf("err :%s\n", err.Error())
			}
		}
	}()
}
